package simpledb.planner;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import simpledb.index.query.IndexJoinPlan;
import simpledb.metadata.IndexInfo;
import simpledb.metadata.MetadataMgr;
import simpledb.parse.QueryData;
import simpledb.query.Constant;
import simpledb.query.Plan;
import simpledb.query.Predicate;
import simpledb.query.ProductPlan;
import simpledb.query.ProjectPlan;
import simpledb.query.SelectPlan;
import simpledb.query.TablePlan;
import simpledb.server.SimpleDB;
import simpledb.tx.Transaction;

/**
 * An optimized query planner. Performs projections and selections as early as
 * possible, heuristically selects join order, and handles nested queries.
 * 
 * @author Timothy Flynn
 * @version December 17, 2011
 */
public class OptQueryPlanner implements QueryPlanner {

	@Override
	public Plan createPlan(QueryData data, Transaction tx) {
		Predicate pred = data.pred();
		MetadataMgr mm = SimpleDB.mdMgr();
		Map<String, IndexInfo> indices = new HashMap<String, IndexInfo>();
		
		// Step 1: Create a list of plans containing only the necessary fields
		// projected, and where each constant selection predicate sits on top 
		// of the table or view it acts on. If no fields are needed, discard
		// the table/view.
		List<Plan> plans = new ArrayList<Plan>();
		
		// For each table/view involved in query
		for(String tblname : data.tables()) {
			Plan p = null;
			boolean needsStepOne = true;

			// Check if we are a table or a view
			String viewdef = mm.getViewDef(tblname, tx);
			if (viewdef != null) { // View
				p = SimpleDB.planner().createQueryPlan(viewdef, tx);
			} else if (tblname.startsWith("tmp")) { // Nested query
				p =  SimpleDB.planner().getTempTablePlan(tblname);
				needsStepOne = false;
			} else { // Table
				p = new TablePlan(tblname, tx);
			}
			
			if(needsStepOne) {
				// Store all of the indices for this table
				Map<String, IndexInfo> indexInfo = mm.getIndexInfo(tblname, tx);
				for(String field : indexInfo.keySet()) {
					indexInfo.get(field).open();
					indices.put(field, indexInfo.get(field));
				}
				// Project only the fields that we need
				Collection<String> fields = new ArrayList<String>();
				
				// Add needed fields from SELECT clause
				for(String s : data.fields()) {
					if(p.schema().hasField(s)) {
						fields.add(s);
					}
				}
				
				// Add needed fields from WHERE clause
				for(String f : p.schema().fields()) {
					Constant c = pred.equatesWithConstant(f);
					String s = pred.equatesWithField(f);
					
					if(c != null || s != null) {
						fields.add(f);
					}
				}
				
				// Project only the fields we need
				if(fields.size() > 0) {
					p = new ProjectPlan(p, fields);
				}
				
				// For each predicate acting on this table
				Predicate ptmp = pred.selectPred(p.schema());
				if(ptmp != null) {
					for(String f : p.schema().fields()) {
						Constant c = ptmp.equatesWithConstant(f);
						String s = ptmp.equatesWithField(f);
						
						// If it is a constant predicate, add selection on top of table
						if(c != null) {
							p = new SelectPlan(p, ptmp);
						}
						
						// If it is a selection predicate, and the rhs is also in this 
						// table, add selection on top of the table
						if(s != null) {
							if(p.schema().hasField(s)) {
								p = new SelectPlan(p, ptmp);
							}
						}
					}
				}
			}

			plans.add(p);
		}

		// Step 2: Create a new list of plans where, for each plan above, selection
		// predicates involving two relations sit on top of the product of those
		// relations. After this, all selections should be complete.
		List<Plan> plans2 = new ArrayList<Plan>();

		// For each plan in the first list
		while(!plans.isEmpty()) {
			Plan p1 = null; //plans.remove(0);
			int minCost = Integer.MAX_VALUE;
			
			// Find lowest cost plan
			for(Plan curr : plans) {
				if(curr.recordsOutput() < minCost) {
					minCost = curr.recordsOutput();
					p1 = curr;
				}
			}
			plans.remove(p1);
			
			// For field in this table
			for(String f : p1.schema().fields()) {
				// If it is a field predicate
				String s = pred.equatesWithField(f);
				if(s != null) {
					
					// Need to find the plan containing rhs
					boolean found = false;
					for(Plan p2 : plans) {
						if(p2.schema().hasField(s)) {
							found = true;
							// Found it! X-product plans and do selection
							Predicate pd = pred.joinPred(p1.schema(), p2.schema());
							if(indices.get(s) != null) {
								p1 = new IndexJoinPlan(p1, p2, indices.get(s), f, tx);
							} else {
								p1 = new ProductPlan(p1, p2);
							}
							p1 = new SelectPlan(p1, pd);
							
							plans.remove(p2);
							break;
						}
					}

					// If not found in old list of plans, it is already in new list
					if(!found) {
						for(Plan p2 : plans2) {
							if(p2.schema().hasField(s)) {
								// Found it! X-product plans and do selection
								Predicate pd = pred.joinPred(p1.schema(), p2.schema());
								if(indices.get(s) != null) {
									p1 = new IndexJoinPlan(p1, p2, indices.get(s), f, tx);
								} else {
									p1 = new ProductPlan(p1, p2);
								}
								p1 = new SelectPlan(p1, pd);

								plans2.remove(p2);
								break;
							}
						}
						
						break;
					}
				}
			}
			
			plans2.add(p1);
		}

		// Step 3: Create the product of all remaining plans
		Plan p = plans2.remove(0);
		for (Plan nextplan : plans2) {
			p = new ProductPlan(p, nextplan);
		}
		
		// Step 4: Project on the field names
		return new ProjectPlan(p, data.fields());

	}
}